package com.bawei.gmall.cancal.client;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.bawei.gmall.cancal.client.utils.KafkaSender;
import com.bawei.gmall.common.GmallConstants;

import java.util.List;

/**
 * @ClassName CanalHanlder
 * @Description TODO
 * @Author mufeng_xky
 * @Date 2020/3/31 11:28
 * @Version V1.0
 **/
public class CanalHanlder {

    /**
     * todo : 定义变量
     * 1. 表名
     * 2. Sql类型
     * 3. 数据集合
     */
    private String tableName;
    private CanalEntry.EventType eventType;
    private List<CanalEntry.RowData> rowDatasList;

    /**
     * 构造器
     *
     * @param tableName
     * @param eventType
     * @param rowDatasList
     */
    public CanalHanlder(String tableName, CanalEntry.EventType eventType, List<CanalEntry.RowData> rowDatasList) {
        this.tableName = tableName;
        this.eventType = eventType;
        this.rowDatasList = rowDatasList;
    }

    /**
     * 执行器
     */
    public void handle() {
        /**
         * 监控订单表 ： 表名与类型匹配
         */
        if (tableName.equals("order_info") && eventType.equals(CanalEntry.EventType.INSERT)) {
            // 循环拿到的数据发送至kafka
            for (CanalEntry.RowData rowData : rowDatasList) {
                sendKafka(rowData, GmallConstants.KAFKA_TOPIC_ORDER);
            }
        }
        /**
         * 监控用户表
         */
        else if (tableName.equals("user_info") &&
                (eventType.equals(CanalEntry.EventType.INSERT)) ||
                (eventType.equals(CanalEntry.EventType.UPDATE))){
            // 循环拿到的数据发送至kafka
            for (CanalEntry.RowData rowData : rowDatasList) {
                sendKafka(rowData, GmallConstants.KAFKA_TOPIC_USER_INFO);
            }
        }
        /**
         * 监控订单详情表
         */
        else if (tableName.equals("order_detail") && (eventType.equals(CanalEntry.EventType.INSERT))){
            // 循环拿到的数据发送至kafka
            for (CanalEntry.RowData rowData : rowDatasList) {
                sendKafka(rowData, GmallConstants.KAFKA_TOPIC_ORDER_DETAIL);
            }
        }
    }

    /**
     * 向kafka发送数据
     *
     * @param rowData
     * @param topic
     */
    private void sendKafka(CanalEntry.RowData rowData, String topic) {
        // 1. 获取数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        // 2. 创建Json容器
        JSONObject jsonObject = new JSONObject();
        // 3. 组合将要发送的结果
        for (CanalEntry.Column column : afterColumnsList) {
            System.out.println(column.getName() + "----->" + column.getValue());
            // 4. 讲发送的数据添加到对应的容器中
            jsonObject.put(column.getName(),column.getValue());
        }
        // 5. 转换为String
        String rowJson = jsonObject.toJSONString();
        // 6. 向kafka发送数据
        KafkaSender.send(topic,rowJson);
    }
}
